package controller

import (
	"bufio"
	log "common/log4go"
	"encoding/base64"
	"encoding/json"
	"errors"
	"io"
	"iot-bridge/iot_cmd_up/distribute/service"
	"os"
	"os/signal"
	"strings"
	"sync"
	"time"

	"github.com/Shopify/sarama"
	"github.com/bsm/sarama-cluster"
)

type DataStruct struct {
	MsgType     string `json:"MsgType"` //no use
	Event       string `json:"Event"`   //no use
	Topic       string `json:"Topic"`
	Seq         int64  `json:"Seq"` //no use
	ProductId   string `json:"ProductId"`
	PayloadLen  int64  `json:"PayloadLen"`  //no use
	DeviceId    string `json:"DeviceId"`    //no use
	ProductName string `json:"ProductName"` //no use
	DeviceName  string `json:"DeviceName"`
	Payload     string `json:"Payload"`
	Time        string `json:"time"` //no use

	ServerTime int64 `json:"serverTime"` //distribute server time

	//aws
	AwsAccountId string `json:"awsAccountId"`
	AwsRequestId string `json:"awsRequestId,omitempty"` //aws request id
	PipeType     string `json:"pipeType"`               //pipe type  2 aws
	AwsMqttTime  int64  `json:"awsMqttTime,omitempty"`  //aws  rule  triggered  time
	AwsLamdaTime int64  `json:"awsLamdaTime"`           //aws lamda fun exec time
	AwsTcpTime   int64  `json:"awsTcpTime"`             //aws tcp time
}

var (
	ErrIllegalHead = errors.New("包头不合法")
	ErrIllegalTail = errors.New("包尾不合法")
	ErrIllegalPack = errors.New("包不合法")
)

// 正式上传的数据包
type ReportPacket struct {
	Content   string `json:"content"`
	Token     string `json:"token"`
	Sn        string `json:"sn"`
	Timestamp int64  `json:"timestamp"`
}

// 支持brokers cluster的消费者
func TcpClusterConsumer(wg *sync.WaitGroup, addrs []string, topics []string, groupId string) {
	//defer wg.Done()
	config := cluster.NewConfig()
	config.Consumer.Return.Errors = true
	config.Group.Return.Notifications = true
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	config.Version = sarama.V0_10_0_0

	// init consumer
	consumer, err := cluster.NewConsumer(addrs, groupId, topics, config)
	if err != nil {
		log.Error("%s: sarama.NewSyncProducer err, message=%s \n", groupId, err)
		return
	}
	defer consumer.Close()
	// trap SIGINT to trigger a shutdown
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)

	// consume errors
	go func() {
		for err := range consumer.Errors() {
			log.Error("%s:Error: %s\n", groupId, err.Error())
		}
	}()

	// consume notifications
	go func() {
		for ntf := range consumer.Notifications() {
			log.Error("%s:Rebalanced: %+v \n", groupId, ntf)
		}
	}()

	// consume messages, watch signals
	var successes int
Loop:
	for {
		select {
		case msg, ok := <-consumer.Messages():
			if ok {
				log.Info("groupid:%s topic:%s partition:%d offset:%d key:%s value:%s", groupId, msg.Topic, msg.Partition, msg.Offset, msg.Key, msg.Value)
				err = ProcessTcpData(msg.Value)
				if err != nil {
					//todo 如果是不合法的包 要单独判断提交读取 避免下次再读
					log.Error("ProcessTcpData(%s) error(%v)", string(msg.Value), err)
					//return
				}
				consumer.MarkOffset(msg, "") // mark message as processed
				log.Info("consumer.MarkOffset(%v, '')", msg.Value)
				successes++
			} else {
				//todo 如果读取失败
			}
		case <-signals:
			break Loop
		}
	}
	log.Error("%s consume %d messages \n", groupId, successes)
}
func ProcessTcpData(value []byte) error {
	var (
		token      string
		statusPack = 0
	)
	data := &DataStruct{}
	err := json.Unmarshal(value, data)
	if err != nil {
		log.Error("json.Unmarshal(%v,%v) error(%v)", value, data, err)
		return err
	}
	data.ServerTime = time.Now().UnixNano()
	if data.Topic == "" {
		log.Info("上下线包丢弃不处理")
		return nil
	}
	//cid := data.DeviceName
	tcpPacket, err := base64.StdEncoding.DecodeString(data.Payload)
	if err != nil {
		log.Error("base64.StdEncoding.DecodeString(%s) error(%v)", tcpPacket, err)
		return err
	}
	//tcpPacket, _ = hex.DecodeString(string(tcpPacket))
	log.Info("tcpPacket(%v)", tcpPacket)
	bufferReader := bufio.NewReader(strings.NewReader(string(tcpPacket)))
	report := []byte{}
	for {
		b := make([]byte, 256)
		n, err := bufferReader.Read(b)
		if err != nil {
			if err == io.EOF || n < 1 {
				log.Error("read kafka pack error")
				return ErrIllegalPack
			}
		}
		// statusPack == 1 时即第一次reader即判断包头合法性
		statusPack++
		if statusPack == 1 {
			if b[0] != 0xEF && b[1] != 0xFE {
				log.Error("包头不合法丢弃")
				return ErrIllegalHead
			}
		}
		report = append(report, b[0:n]...)
		if n < 256 {
			break
		}
	}

	if len(report) <= 6 {
		log.Error("包长不合法")
		return ErrIllegalPack
	}

	//包剩余长度

	//计算包剩余长度
	pack_len, count := packetLen(report[2:])
	if int(pack_len+count+2) > len(report) {
		log.Error("包长不合法 pack_len(%d) count(%d) report(%d)", pack_len, count, len(report))
		return ErrIllegalPack
	}

	if pack_len <= 2 {
		log.Error("bad pack bbbbbbbbbbbbbbbb  data(%s)", string(value))
		return nil
	}

	//去掉包头+包剩余长度的包
	if 2+int(count) > len(report) {
		log.Error("包长不合法 pack(%v)", value)
		return ErrIllegalPack
	}
	packWithBottom := report[2+int(count):]
	if int(pack_len) > len(packWithBottom) {
		log.Error("包长不合法")
		return ErrIllegalPack
	}
	//判断包尾
	bottomPackLen := len(packWithBottom)
	if bottomPackLen < 2 {
		log.Error("包长不合法 pack(%v)", value)
		return ErrIllegalPack
	}
	if packWithBottom[bottomPackLen-2] != 0xFA && packWithBottom[bottomPackLen-1] != 0xEA {
		log.Error("包不合法丢弃")
		//fmt.Println("包不合法丢弃")
		return ErrIllegalPack
	}

	cidLen := int(packWithBottom[0] & 127)

	if cidLen+1 > len(packWithBottom) {
		log.Error("illegal package(%v)", packWithBottom)
		return ErrIllegalPack
	}

	cid := string(packWithBottom[1 : cidLen+1])
	if cid != data.DeviceName {
		log.Error("设备cid不同 pack[0](%v) cidlen(%d) cid(%s) deviceName(%v)", packWithBottom[0], cidLen, cid, data.DeviceName)
		return ErrIllegalPack
	}
	payloadStart := 1 + cidLen

	if payloadStart+16 > len(packWithBottom) {
		log.Error("illegal package(%v)", packWithBottom)
		return ErrIllegalPack
	}
	if packWithBottom[0]&(1<<7) != 0 {
		token = string(packWithBottom[payloadStart : payloadStart+16])
		payloadStart += 16
	}
	//指令后续处理
	log.Info("packWithBottom(%v) payloadStart(%d) end(%d)", packWithBottom, payloadStart, pack_len-2)

	if int(pack_len-2) > len(packWithBottom) {
		log.Error("bad pack aaaaaaaaaaaaaaaaaaaaaaaa   value(%s)", string(value))
		return nil
	}

	payload := packWithBottom[payloadStart : pack_len-2]
	go func() {
		err = TcpReportService.DataToKafka(cid, payload)
		if err != nil {
			log.Error("TcpReportService.DataToKafka(%s, %v) error(%v)", cid, payload, err)
			return
		}
		// 记录最后一次上报到Redis中
		err := service.SaveIotUpTime(cid)
		if err != nil {
			log.Error("service.SaveIotUpTime(%s) error(%v)", cid, err)
			return
		}
	}()
	//ToDo 封装拆包
	sn := ""
	go func() {
		//fmt.Printf("token(%s) table(%s) tcpReport cid(%s) sn(%s) payload(%s) productid(%s)\n", token, service.MakeTable(data.ProductId), cid, sn, string(payload))
		err = TcpReportService.DataToDb(token, service.MakeTable(data.ProductId), TcpReport, cid, sn, report)
		if err != nil {
			log.Error("TcpReportService.DataToDb(%s, %s, %d, %s, %s, %s) error(%v)", token, service.MakeTable(data.ProductId), TcpReport, cid, sn, string(report), err)
			//todo 判断返回错误
			return
		}
	}()
	return nil
}
func ProcessTcpDataIOTSAAS(value []byte) error {
	var (
		token      string
		statusPack = 0
	)
	data := &DataStruct{}
	data.Payload = string(value)
	data.ServerTime = time.Now().UnixNano()
	if data.Topic == "" {
		log.Info("上下线包丢弃不处理")
		return nil
	}
	//cid := data.DeviceName
	tcpPacket, err := base64.StdEncoding.DecodeString(data.Payload)
	if err != nil {
		log.Error("base64.StdEncoding.DecodeString(%s) error(%v)", tcpPacket, err)
		return err
	}
	//tcpPacket, _ = hex.DecodeString(string(tcpPacket))
	log.Info("tcpPacket(%v)", tcpPacket)
	bufferReader := bufio.NewReader(strings.NewReader(string(tcpPacket)))
	report := []byte{}
	for {
		b := make([]byte, 256)
		n, err := bufferReader.Read(b)
		if err != nil {
			if err == io.EOF || n < 1 {
				log.Error("read kafka pack error")
				return ErrIllegalPack
			}
		}
		// statusPack == 1 时即第一次reader即判断包头合法性
		statusPack++
		if statusPack == 1 {
			if b[0] != 0xEF && b[1] != 0xFE {
				log.Error("包头不合法丢弃")
				return ErrIllegalHead
			}
		}
		report = append(report, b[0:n]...)
		if n < 256 {
			break
		}
	}

	if len(report) <= 6 {
		log.Error("包长不合法")
		return ErrIllegalPack
	}

	//包剩余长度

	//计算包剩余长度
	pack_len, count := packetLen(report[2:])
	if int(pack_len+count+2) > len(report) {
		log.Error("包长不合法 pack_len(%d) count(%d) report(%d)", pack_len, count, len(report))
		return ErrIllegalPack
	}

	if pack_len <= 2 {
		log.Error("bad pack bbbbbbbbbbbbbbbb  data(%s)", string(value))
		return nil
	}

	//去掉包头+包剩余长度的包
	if 2+int(count) > len(report) {
		log.Error("包长不合法 pack(%v)", value)
		return ErrIllegalPack
	}
	packWithBottom := report[2+int(count):]
	if int(pack_len) > len(packWithBottom) {
		log.Error("包长不合法")
		return ErrIllegalPack
	}
	//判断包尾
	bottomPackLen := len(packWithBottom)
	if bottomPackLen < 2 {
		log.Error("包长不合法 pack(%v)", value)
		return ErrIllegalPack
	}
	if packWithBottom[bottomPackLen-2] != 0xFA && packWithBottom[bottomPackLen-1] != 0xEA {
		log.Error("包不合法丢弃")
		//fmt.Println("包不合法丢弃")
		return ErrIllegalPack
	}

	cidLen := int(packWithBottom[0] & 127)

	if cidLen+1 > len(packWithBottom) {
		log.Error("illegal package(%v)", packWithBottom)
		return ErrIllegalPack
	}

	cid := string(packWithBottom[1 : cidLen+1])
	if cid != data.DeviceName {
		log.Error("设备cid不同 pack[0](%v) cidlen(%d) cid(%s) deviceName(%v)", packWithBottom[0], cidLen, cid, data.DeviceName)
		return ErrIllegalPack
	}
	payloadStart := 1 + cidLen

	if payloadStart+16 > len(packWithBottom) {
		log.Error("illegal package(%v)", packWithBottom)
		return ErrIllegalPack
	}
	if packWithBottom[0]&(1<<7) != 0 {
		token = string(packWithBottom[payloadStart : payloadStart+16])
		payloadStart += 16
	}
	//指令后续处理
	log.Info("packWithBottom(%v) payloadStart(%d) end(%d)", packWithBottom, payloadStart, pack_len-2)

	if int(pack_len-2) > len(packWithBottom) {
		log.Error("bad pack aaaaaaaaaaaaaaaaaaaaaaaa   value(%s)", string(value))
		return nil
	}

	payload := packWithBottom[payloadStart : pack_len-2]
	go func() {
		err = TcpReportService.DataToKafka(cid, payload)
		if err != nil {
			log.Error("TcpReportService.DataToKafka(%s, %v) error(%v)", cid, payload, err)
			return
		}
		// 记录最后一次上报到Redis中
		err := service.SaveIotUpTime(cid)
		if err != nil {
			log.Error("service.SaveIotUpTime(%s) error(%v)", cid, err)
			return
		}
	}()
	//ToDo 封装拆包
	sn := ""
	go func() {
		//fmt.Printf("token(%s) table(%s) tcpReport cid(%s) sn(%s) payload(%s) productid(%s)\n", token, service.MakeTable(data.ProductId), cid, sn, string(payload))
		err = TcpReportService.DataToDb(token, service.MakeTable(data.ProductId), TcpReport, cid, sn, report)
		if err != nil {
			log.Error("TcpReportService.DataToDb(%s, %s, %d, %s, %s, %s) error(%v)", token, service.MakeTable(data.ProductId), TcpReport, cid, sn, string(report), err)
			//todo 判断返回错误
			return
		}
	}()
	return nil
}
